package com.bml.architect.order;

import com.bml.architect.utils.MqConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 顺序消息的消费
 * 然后在消费者端：消费者会从多个消息队列上去拿消息。这时虽然每个消息队列上的消息是有序
 * 的，但是多个队列之间的消息仍然是乱序的。消费者端要保证消息有序，就需要按队列一个一个
 * 来取消息，即取完一个队列的消息后，再去取下一个队列的消息。而给consumer注入的
 * MessageListenerOrderly对象，在RocketMQ内部就会通过锁队列的方式保证消息是一个一个队
 * 列来取的。MessageListenerConcurrently这个消息监听器则不会锁队列，每次都是从多个
 * Message中取一批数据（默认不超过32条）。因此也无法保证消息有序。
 */
public class Consumer {

    public static void main(String[] args) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderConsumerGroup");

        try {
            consumer.setNamesrvAddr(MqConstants.getNameServerAddr());
            consumer.subscribe("OrderTopicTest", "*");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                    consumeOrderlyContext.setAutoCommit(true);
                    for (MessageExt msg : list) {
                        System.out.println("收到消息内容 " + new String(msg.getBody()));
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }


    }
}
